package com.mfs.proxy.core.server.reactor;

import com.alibaba.fastjson2.JSON;
import com.mfs.common.constant.Constant;
import com.mfs.common.utills.ChannelUtil;
import com.mfs.common.utills.ObjectUtil;
import com.mfs.proxy.core.exceptions.ProxyException;
import com.mfs.proxy.core.enums.MessageTypeEnum;
import com.mfs.proxy.core.message.Message;
import com.mfs.proxy.core.server.endpoint.ProxyHandle;
import com.mfs.proxy.core.server.registry.ProxyClientRegistrySingletonFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

public class SubServerReactor implements Runnable {
    private final static Logger logger = LoggerFactory.getLogger(SubServerReactor.class);
    private static Selector subServerSelector;
    public SubServerReactor() {
        try {
            subServerSelector = Selector.open();
        } catch (IOException e) {
            throw new ProxyException(e);
        }
    }

    public static SelectionKey register(SelectableChannel selectableChannel) throws ClosedChannelException {
        return selectableChannel.register(subServerSelector, SelectionKey.OP_READ);
    }

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                int select = subServerSelector.select(100);
                if (select < 1) {
                    Thread.yield();
                    continue;
                }
                Set<SelectionKey> selectionKeys = subServerSelector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    processReadable(selectionKey);
                    iterator.remove();
                }
            } catch (Exception e) {
                logger.error("处理客户端消息失败", e);
            }
        }
    }

    /**
     * 仅处理readable
     * 处理客户端消息，包括所有{@link com.mfs.proxy.core.enums.MessageTypeEnum} 中，代理客户端为发送主体的消息
     * */
    private void processReadable(SelectionKey selectionKey) {
        if (!(selectionKey.isValid() && selectionKey.isReadable())) {
            return;
        }
        // 1. 获取message对象
        SocketChannel clientChannel = (SocketChannel) selectionKey.channel();
        byte[] clientMsg = new byte[0];
        try {
            clientMsg = ChannelUtil.read(clientChannel);
        } catch (IOException e) {
            logger.error("读取信息失败", e);
            selectionKey.cancel();
            return;
        }
        if (clientMsg.length < Constant.ONE) {
            // 客户端断开连接
            try {
                clientChannel.close();
            } catch (IOException e) {
                logger.error("连接已关闭", e);
            }
            selectionKey.cancel();
            return;
        }
        Message message = ObjectUtil.deserialize(clientMsg, Message.class);

        // 2. 匹配消息枚举
        MessageTypeEnum messageEnum = MessageTypeEnum.valueOf(message.getMessageType());
        if (messageEnum == null) {
            logger.error("不能处理的消息类型:{}", JSON.toJSONString(message));
            return;
        }

        // 3.消息处理
        Object messageContent = message.getMessageContent();
        try {
            logger.debug("正在处理客户端消息，消息类型：{}，消息内容：{}", messageEnum.getDescribe(), JSON.toJSONString(message.getMessageContent()));
            messageEnum.getMessageReceiveProcessor().receive(selectionKey, messageContent);
        } catch (IOException e) {
            logger.error("处理客户端消息发送异常", e);
            try {
                selectionKey.channel().close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
            selectionKey.cancel();
            if (messageContent instanceof ProxyHandle) {
                ProxyHandle proxyHandle = (ProxyHandle) messageContent;
                ProxyClientRegistrySingletonFactory.getProxyClientRegistry().unregister(proxyHandle.getProxyIp(), proxyHandle.getProxyPort());
            }
        }

    }
}
